package com.idroc.rpc;

import java.io.InputStream;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.apache.mina.common.ByteBuffer;
import org.apache.mina.common.IoSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.caucho.hessian.client.HessianRuntimeException;
import com.caucho.hessian.io.AbstractHessianInput;
import com.caucho.hessian.io.AbstractHessianOutput;
import com.caucho.hessian.io.Hessian2Input;
import com.caucho.hessian.io.HessianOutput;
import com.caucho.hessian.io.HessianProtocolException;
import com.caucho.hessian.io.HessianRemoteObject;
import com.caucho.hessian.io.SerializerFactory;
import com.caucho.services.server.AbstractSkeleton;
import com.idroc.message.Message;
import com.idroc.message.MessageHandler;
import com.idroc.message.MessageService;
import com.idroc.message.impl.MessageClient;

/**
 * rpc服务提供者
 * @author 赤脚詹(yuyoo4j@163.com)
 * @date Aug 17, 2009 4:33:29 PM 
 * @memo:
 */
public class RPCService {
	
	
	private InetSocketAddress endpoint; 
	private MessageClient client = null;
	
	private Map<UUID, BlockingQueue<RPCMessage>> requests = new ConcurrentHashMap<UUID, BlockingQueue<RPCMessage>>(1500, 0.8f, 50);
		
	RPCService(InetSocketAddress endpoint) {
		
		this.endpoint = endpoint;
		Map<String, String> config = new HashMap<String, String>();
		config.put(MessageClient.CONNECT_TO_HOST, endpoint.getHostName());
		config.put(MessageClient.CONNECT_TO_PORT, Integer.toString(endpoint.getPort()));
		MessageHandler handler = new InnerMsgHandler();
		client = new MessageClient();
		client.init(config);
		client.registerMsgHandler(RPCMessage.class, handler);
		client.start(); 
	}

	/**
	 * 获取rpc服务代理
	 * @param itfc -- 接口类
	 * @return -- 接口类的代理
	 */
	@SuppressWarnings("unchecked")
	public <T> T importInstance(Class itfc) {
		
		HessianProxy proxy = new HessianProxy(itfc, endpoint);
		ClassLoader cl = Thread.currentThread().getContextClassLoader();
		return (T) Proxy.newProxyInstance(cl,
                					new Class[] { itfc, HessianRemoteObject.class },
                					proxy); 
	} 
	
	/**
	 * 
	 * @author 赤脚詹(yuyoo4j@163.com)
	 * @date Aug 17, 2009 4:33:59 PM 
	 * @memo:
	 */
	@SuppressWarnings("unchecked")
	private class HessianProxy implements InvocationHandler {
		
		private final Logger log = LoggerFactory.getLogger(this.getClass());
		
		private Map<Method, String> cacheMethods = new ConcurrentHashMap<Method, String>(300);
		
		private InetSocketAddress endpoint = null;
		private Class itfc = null;
		private int serviceId = -1;
		
		private SerializerFactory serializer = new SerializerFactory(); 
		
		HessianProxy(Class itfc, InetSocketAddress ep) {
			this.endpoint = ep;	
			this.itfc = itfc;
			this.serviceId = this.itfc.getName().hashCode();
		}

		public Object invoke(Object proxy, Method method, Object[] args)
				throws Throwable { 
			
			// 处理rpc请求	
			String methodName = cacheMethods.get(method);
			if (null == methodName) {
				
				String mn = method.getName();
				Class []params = method.getParameterTypes();
				
				if ("equals".equals(mn) && params.length == 1 && params[0].equals(Object.class)) {
					
					Object value = args[0];
					if (null == value || ! Proxy.isProxyClass(value.getClass())) {
						return Boolean.FALSE;
					}
					Object proxyHandler = Proxy.getInvocationHandler(value);
					if (! (proxyHandler instanceof HessianProxy)) {
						  return Boolean.FALSE;
					}
					HessianProxy handler = (HessianProxy) proxyHandler;
					return new Boolean(RPCService.this.endpoint.equals(handler.endpoint));
				} else if ("hashCode".equals(mn) && params.length == 0) {
					
					return new Integer(endpoint.hashCode());
				} else if ("getHessianType".equals(mn)) {
					return proxy.getClass().getInterfaces()[0].getName();
				} else if ("toString".equals(mn) && params.length == 0) {
					return "RPCService.HessianProxy[" + endpoint.toString() + "]";
				}
				methodName = mangleName(method);
				cacheMethods.put(method, methodName);
			}
			ByteBuffer data = ByteBuffer.allocate(512).setAutoExpand(true);
			AbstractHessianOutput output = new HessianOutput(data.asOutputStream());
			output.setSerializerFactory(serializer);
			output.call(methodName, args);
			output.flush();
			output.close();
			RPCMessage msg = new RPCMessage(); 
			msg.setRequestData(data);
			msg.setServiceId(serviceId);
			UUID reqId = UUID.randomUUID();
			msg.setRequestId(reqId);
			 		
			BlockingQueue<RPCMessage> res = new LinkedBlockingQueue<RPCMessage>(1);
			requests.put(reqId, res);			
			client.sendMessage(msg);
			
			log.debug("发送RPC请求消息:{}", msg);
			
			// 处理rpc响应	
			try {
				RPCMessage resMsg = res.poll(60, TimeUnit.SECONDS);
				log.debug("接受到RPC响应消息:{}", resMsg);
				
				ByteBuffer resData = resMsg.getRequestData();
				int code = resData.get();
				AbstractHessianInput in = null;
				if (code == 'H') {
					int major = resData.get();
					int minor = resData.get();
					in = new Hessian2Input(resData.asInputStream());
					return in.readReply(method.getReturnType());
				} else if (code == 'r') {
					in = new Hessian2Input(resData.asInputStream());
					in.setSerializerFactory(serializer); 
					in.startReply();
					Object value = in.readObject(method.getReturnType());
					if (value instanceof InputStream) {
						 return value;
					} else {
						in.completeReply(); 
						return value;
					}
				} else {
					throw new HessianProtocolException("'" + (char) code + "' is an unknown code");
				}
			} catch (Exception ex) {
				log.error("请求ID(reqId):{}", reqId);
				log.error("请求消息(RPCMessage):{}", msg);
				log.error("等待请求响应异常", ex); 
				throw new HessianRuntimeException("rpc调用异常", ex);
			} finally {
				requests.remove(reqId);
			}			 
		}
		
		/**
		 * 辗碎方法,将方法变成一个string
		 * @param method -- 方法
		 * @return -- 方法对应的辗碎后的字符串
		 */
		private String mangleName(Method method) {
			
		    Class []param = method.getParameterTypes();
		    
		    if (param == null || param.length == 0) {
		      return method.getName();
		    } else {
		      return AbstractSkeleton.mangleName(method, false);
		    }
	    }
		
	}
	
	/**
	 * 消息处理器
	 * @author 赤脚詹(yuyoo4j@163.com)
	 * @date Aug 17, 2009 4:34:12 PM 
	 * @memo:
	 */
	private class InnerMsgHandler implements MessageHandler {
		
		private final Logger log = LoggerFactory.getLogger(this.getClass());

		public void onAttached(MessageService service) { 
			
			log.debug("service:{}", service);
		}

		public void onMsgCast(IoSession session, Message message) { 

			log.debug("message:{}", message);
		}

		public void onMsgReceived(IoSession session, Message message) { 
			
			RPCMessage msg = (RPCMessage) message;
			UUID reqId = msg.getRequestId();
			log.debug("请求ID:{}, 消息响应:{}", reqId, msg);
			BlockingQueue res = requests.get(reqId);
			if (null != res) {
				try { 
					res.add(msg);
				} catch (Exception ex) {
					log.error("请求ID:{}, 消息响应:{}", reqId, msg);
					log.error("无法将消息响应放入请求队列", ex);
				}
			} else {
				log.error("请求ID:{}, 消息响应:{}", reqId, msg);
				log.error("没有找到对应请求队列");
			}
		}

		public void onMsgSent(IoSession session, Message message) { 

			log.debug("message:{}", message);
		}

		public void onReleased(MessageService service) { 

			log.debug("service:{}", service);
		}		
	}
}
